package com.atguigu.flink.streamapi.sink;

import com.alibaba.fastjson.JSON;
import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;
import java.nio.charset.StandardCharsets;
import java.util.Properties;

/**
 * Created by Smexy on 2023/1/30
 *      写出的数据应该K-V结构。
 *             第一种： 不需要指定key
 *                          生产者默认用轮询的方式分区，保证负载均衡！
 *             第二种:  需要指定key
 *                          key用来分区，key相同，一定写入kafka的同一个分区！
 *
 *     -----------------
 *      本质就是构造一个生产者，在flink中称为 FlinkKafkaProducer
 */
public class Demo3_KafkaSinkWithKey
{
    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        Properties properties = new Properties();
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop103:9092");

        /*
            第二种:  需要指定key
             FlinkKafkaProducer(
            String defaultTopic,    // 没有作用，随便写
            KafkaSerializationSchema<IN> serializationSchema // 指定如何把流中的一条数据，序列化(详细明确，key怎么序列化，value怎么序列化)
                IN: 被序列化的value的类型，也就是上游发往sink的类型
            Properties producerConfig  // 生产者的配置信息
            FlinkKafkaProducer.Semantic semantic  //语义，可以结合flink的容错机制保障 语义正确
            )
         */
        FlinkKafkaProducer kafkaProducer = new FlinkKafkaProducer(
            "无",
            new KafkaSerializationSchema<WaterSensor>(){

                /*
                     把WaterSensor 封装为 ProducerRecord

                     ProducerRecord 是 key-value！
                 */
                @Override
                public ProducerRecord<byte[], byte[]> serialize(WaterSensor element, @Nullable Long timestamp) {
                    //生成key 按照传感器的id作为key，同一种传感器分到同一个partition
                    byte[] key = element.getId().getBytes(StandardCharsets.UTF_8);

                    //value： 数据本身  为了在客户端查看方便。 对象 ----> jsonstr ---->byte[]
                    byte[] value = JSON.toJSONString(element).getBytes(StandardCharsets.UTF_8);

                    return new ProducerRecord<>("topicB",key,value);
                }
            },
            properties,
            FlinkKafkaProducer.Semantic.AT_LEAST_ONCE
        );

        env
                   .socketTextStream("hadoop103", 8888)
                   .map(new WaterSensorMapFunction())
                   .addSink(kafkaProducer);


                try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }

    }
}
